package com.yuwen.yuwenRPC.registry;

/**
 * @ClassNameEtcdRegistry
 * @Description TODO
 * @Author钰玟
 * @Date2025/3/31 下午5:17
 * @Version 1.0
 **/

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.ConcurrentHashSet;
import cn.hutool.cron.CronUtil;
import cn.hutool.cron.task.Task;
import cn.hutool.json.JSONUtil;
import com.yuwen.yuwenRPC.config.RegistryConfig;
import com.yuwen.yuwenRPC.model.ServiceMetaInfo;
import io.etcd.jetcd.*;
import io.etcd.jetcd.kv.DeleteResponse;
import io.etcd.jetcd.kv.PutResponse;
import io.etcd.jetcd.options.GetOption;
import io.etcd.jetcd.options.PutOption;
import io.etcd.jetcd.watch.WatchEvent;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

/**
 * etcd 注册中心
 */
@Slf4j
public class EtcdRegistry implements Registry{
    /**
     * etcd根路径
     */
    private final String ETCD_ROOT_PATH = "/rpc/";
    /**
     * etcd 客户端服务
     */
    private Client client;
    /**
     * kvClient：用于对etcd 中的键值对进行操作。通过kvClient 可以进行设置值、获取值、删除值、列出目录等操作。
     */
    private KV kvClient;

    /**
     * 本机注册的 服务节点 集合 (用于维护续期)  存储服务节点的  key
     */
    private final Set<String> localRegisterNodeKeySet = new HashSet<>();

    /**
     * 注册中心本地缓存
     */
    @Deprecated
    private final RegistryServiceCache registryServiceCache = new RegistryServiceCache();

    /**
     * 注册中心本地缓存 支持多服务
     */
    private final RegistryServiceMultiCache registryServiceMultiCache = new RegistryServiceMultiCache();

    /**
     * ConcurrentHashSet
     */
    private final Set<String> watchingKey = new ConcurrentHashSet<>();


    /**
     * 初始化init
     * @param registryConfig 配置
     */
    @Override
    public void init(RegistryConfig registryConfig) {
        client = Client.builder()
                .endpoints(registryConfig.getRegistryAddress())//地址
                .connectTimeout(Duration.ofMillis(registryConfig.getTimeout()))//请求时间
                .build();
        //获取kv
        kvClient = client.getKVClient();

        /**
         * 开启心跳检测方法
         */
        heartBeat();
    }

    /**
     * 注册服务
     * @param serviceMetaInfo 注册服务的服务元信息
     * @throws Exception
     */
    @Override
    public void register(ServiceMetaInfo serviceMetaInfo) throws Exception {
        //1.获取租约client
        Lease lease = client.getLeaseClient();

        //2. 封装要write的信息
        //2.1 key
        String serviceKey = ETCD_ROOT_PATH + serviceMetaInfo.getServiceNodeKey();
        ByteSequence key = ByteSequence.from(serviceKey, StandardCharsets.UTF_8);
        //2.2 value
        ByteSequence value = ByteSequence.from(JSONUtil.toJsonStr(serviceMetaInfo), StandardCharsets.UTF_8);

        //3.将租约和kv关联起来
        //  3.1 设置租约时间
        long leaseId = lease.grant(30).get().getID();
        //  3.2设置 PutOption
        PutOption putOption = PutOption
                                .builder()
                                .withLeaseId(leaseId)
                                .build();
        //4. 放入kvClient
        kvClient.put(key, value, putOption).get();


        //5.放入 服务节点集合
        localRegisterNodeKeySet.add(serviceKey);
    }

    /**
     * 注销服务
     * @param serviceMetaInfo 对应服务元信息
     */
    @Override
    public void unRegister(ServiceMetaInfo serviceMetaInfo) throws ExecutionException, InterruptedException {
        String serviceKey = ETCD_ROOT_PATH + serviceMetaInfo.getServiceNodeKey();
        kvClient.delete(ByteSequence.from(serviceKey, StandardCharsets.UTF_8)).get();

        //移除服务节点集合中的服务
        localRegisterNodeKeySet.remove(serviceKey);
    }

    /**
     * 发现服务
     * @param serviceKey the key service of discovery
     * @return
     */
    @Override
    public List<ServiceMetaInfo> registryDiscovery(String serviceKey) {
        /**
         * 优先从本地缓存获取 服务
         */
//        @Deprecated
//        List<ServiceMetaInfo> serviceMetaInfoCache = registryServiceCache.readServiceCache();

        List<ServiceMetaInfo> serviceMetaInfoCache = registryServiceMultiCache.readServiceCache(serviceKey);
        if (!CollUtil.isEmpty(serviceMetaInfoCache)){
            return serviceMetaInfoCache;
        }


        //层级关系 不能忽略 "/"
        //1. 搜索前缀
        String searchPrefix = ETCD_ROOT_PATH + serviceKey + "/";
        //2.获得对应前缀的服务
        //  2.1 开启前缀搜索
        GetOption getOption = GetOption
                            .builder()
                            .isPrefix(true)
                            .build();
        //  2.2 搜索
        try {
            //将ByteSequence 通过流转化为 我们需要的类型
            List<KeyValue> keyValues = kvClient.get(ByteSequence.from(searchPrefix, StandardCharsets.UTF_8), getOption).get().getKvs();

            List<ServiceMetaInfo> serviceMetaInfoList =
                    keyValues.stream()
                    .map(keyValue -> {
                        /**
                         * 获取key
                         */
                        String key = keyValue.getKey().toString(StandardCharsets.UTF_8);
                        //监视key
                        watch(key);

                        String value = keyValue.getValue().toString(StandardCharsets.UTF_8);
                        return JSONUtil.toBean(value, ServiceMetaInfo.class);
                    }).collect(Collectors.toList());
            /**
             * 写入缓存Set
             */
            registryServiceMultiCache.writeCache(serviceKey,serviceMetaInfoList);
            return serviceMetaInfoList;
        } catch (Exception e) {
            throw new RuntimeException("获取服务列表失败",e);
        }
    }

    @Override
    public void destroy() {
        log.info("当前节点下线");

        for (String key : localRegisterNodeKeySet) {
            try {
                kvClient.delete(ByteSequence.from(key,StandardCharsets.UTF_8)).get();
            } catch (Exception e) {
                throw new RuntimeException(key+"当前节点下线失败",e);
            }
        }
        if(kvClient != null) {
            kvClient.close();
        }
        if(client != null) {
            client.close();
        }
    }

    /**
     * 心跳检测方法
     */
    @Override
    public void heartBeat() {
        //利用hutool工具包的 定时任务*/10 * * * * *
        CronUtil.schedule("*/10 * * * * *", new Task() {
            @Override
            public void execute() {
                //1. 遍历服务节点集合
                for (String key : localRegisterNodeKeySet) {
                    try {
                        //2. 根据key 获取对应的服务
                        List<KeyValue> keyValues = kvClient.get(ByteSequence.from(key, StandardCharsets.UTF_8)).get().getKvs();
                        if(CollUtil.isEmpty(keyValues)){
                            //如果不是空
                            continue;
                        }
                        KeyValue keyValue = keyValues.get(0);
                        //3. 进行续期 操作  即 重新注册
                        String value = keyValue.getValue().toString(StandardCharsets.UTF_8);
                        //Json 转化为Json
                        ServiceMetaInfo serviceMetaInfo = JSONUtil.toBean(value, ServiceMetaInfo.class);
                        //进行续期
                        register(serviceMetaInfo);
                    } catch (Exception e) {
                        throw new RuntimeException(key+"续期失败",e);
                    }
                }
            }

        });
        //开启毫秒级别的任务
        CronUtil.setMatchSecond(true);
        CronUtil.start();
    }


    /**
     * 监听 key的  变化
     * @param serviceNodeKey 所监听的key
     */
    @Override
    public void watch(String serviceNodeKey) {
        //获取监听 客户端
        Watch watchClient = client.getWatchClient();

        //之前未被监听  ，开启监听
        boolean newWatch = watchingKey.add(serviceNodeKey);
        if(newWatch){
            watchClient.watch(ByteSequence.from(serviceNodeKey,StandardCharsets.UTF_8),response ->{
                for (WatchEvent event : response.getEvents()) {
                    switch (event.getEventType()) {
                        //key删除时触发
                        case DELETE:
                            //清理缓存服务
                            /**
                             * serviceNodeKey = > serviceName+serviceVersion/serviceHost:servicePort ---error
                             * serviceKey => serviceName+serviceVersion  ---right
                             */
                            String serviceKey = serviceNodeKey.substring(0, serviceNodeKey.indexOf('/'));
//                            registryServiceCache.clearCache();
                            registryServiceMultiCache.clearCache(serviceKey);
                            break;
                        case PUT:
                        default:
                            break;
                    }
                }
            });
        }


    }
}